package cn.com.lyb.flink.source;

import com.ververica.cdc.connectors.postgres.PostgreSQLSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class PgSourceDemo {
    private volatile boolean isRunning = true;

    // PostgreSQL 配置
    private static final String JDBC_URL = "jdbc:postgresql://172.32.1.6:5432/alone_xgov?currentSchema=aaa";
    private static final String USERNAME = "postgres";
    private static final String PASSWORD = "postgres";

    // 查询语句
    private static final String QUERY = "SELECT * FROM t1";
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 如何将pg信息录入
        env.addSource(new SourceFunction<Object>() {
            @Override
            public void run(SourceContext<Object> ctx) throws Exception {
                Connection conn = DriverManager.getConnection(JDBC_URL, USERNAME, PASSWORD);
                PreparedStatement stmt = conn.prepareStatement(QUERY);
                ResultSet rs = stmt.executeQuery();
                while (rs.next()) {
                    // 假设返回的每一行数据是一个用户
                    int id = rs.getInt("id");
                    String name = rs.getString("name");
                    // 将数据发送到下游（可以是某个流）
                    ctx.collect("User ID: " + id + ", Name: " + name);
                }
                rs.close();
                stmt.close();
                conn.close();
            }

            @Override
            public void cancel() {

            }
        }).print();
        env.execute();
    }

}
